package mqtt;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.jmeter.engine.util.NoThreadClone;
import org.apache.jmeter.samplers.Remoteable;
import org.apache.jmeter.samplers.SampleEvent;
import org.apache.jmeter.samplers.SampleListener;
import org.apache.jmeter.samplers.SampleResult;
import org.apache.jmeter.testelement.AbstractTestElement;
import org.apache.jmeter.testelement.TestStateListener;
import org.apache.jmeter.testelement.ThreadListener;
import org.apache.jmeter.util.JMeterUtils;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.Serializable;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.Properties;

/**
 * 单线程的生产者
 * 优点是确保写入的消息是顺序的
 */
public class MqttSummary extends AbstractTestElement
		implements Serializable, SampleListener, TestStateListener, Remoteable, ThreadListener, NoThreadClone {

	private static final Logger log = LoggerFactory.getLogger(MqttSummary.class);
	private static final long serialVersionUID = 1995L;
	private static final String KAFKA_HOST = JMeterUtils.getPropDefault("KAFKA_HOST", "127.0.0.1:9092");
	int i = 0;
	private String TOPIC_NAME;
	private Properties props = new Properties();
	private Producer<String, String> producer = null;
	private String host;

	{
		props.put("bootstrap.servers", KAFKA_HOST);
		props.put("acks", "0");
		props.put("retries", "0");
		props.put("batch.size", "16384");
		props.put("linger.ms", "1");
		props.put("buffer.memory", "33554432");
		props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
		props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
	}

	{
		try {
			host = InetAddress.getLocalHost().getHostAddress();
		} catch (UnknownHostException e) {
			log.error(e.getMessage(), e);
		}
	}

	public MqttSummary() {
		super();
	}

	public MqttSummary(String name) {
		this();
		setName(name);
	}

	public static void main(String[] args) {

		MqttSummary summary = new MqttSummary();
		System.out.println(summary.host);
	}

	@Override
	public void threadFinished() {

	}

	@Override
	public void threadStarted() {

	}

	@Override
	public void testEnded() {
		producer.close();
	}

	@Override
	public void testEnded(String string) {
		testEnded();
	}

	@Override
	public void testStarted() {

		// 设置一下topic名称
		// this.TOPIC_NAME = FileServer.getFileServer().getScriptName().replace(".jmx", "");
		this.TOPIC_NAME = "test";
		log.info("set TOPIC_NAME: " + TOPIC_NAME);

		// 创建一个生产者对象
		producer = new KafkaProducer<String, String>(props);
		log.info("create kafka producer");
	}

	@Override
	public void testStarted(String string) {
		testStarted();
	}

	// TODO 收集采样结果，发送到Redis消息队列中
	@Override
	public void sampleOccurred(SampleEvent event) {

		long current = System.currentTimeMillis();

		SampleResult result = event.getResult();

		if (result.getSampleLabel().equals("MQTT Sub Sampler")) {

			JSONObject responseJson = JSON.parseObject(result.getResponseDataAsString());

			// elapsed,success,failureMessage,bytes,sentBytes,allThreads,host
			// 收集采样结果
			JSONObject resultJson = new JSONObject();
//			resultJson.put("elapsed", (current - responseJson.getLong("timeStamp")));
			// 只能计算这个时间，因为不同服务器之间的时间太不准了
			resultJson.put("elapsed", (current - responseJson.getJSONObject("data").getLong("timeStamp")));
			if (result.isSuccessful()) {
				resultJson.put("success", true);
				resultJson.put("failureMessage", "");
			} else {
				resultJson.put("success", false);
				resultJson.put("failureMessage", result.getResponseDataAsString());
			}
			resultJson.put("bytes", result.getBytesAsLong());
			resultJson.put("sentBytes", result.getSentBytes());
			resultJson.put("allThreads", result.getAllThreads());
			resultJson.put("host", host);

			// 发送消息
			producer.send(new ProducerRecord<String, String>(TOPIC_NAME, result.getSampleLabel(), resultJson.toJSONString()));
//			log.info("发送消息");
		}
	}

	@Override
	public void sampleStarted(SampleEvent event) {

	}

	@Override
	public void sampleStopped(SampleEvent event) {

	}

}
